package com.niit.sequenceFile.write;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.SequenceFile;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.mapreduce.lib.output.SequenceFileOutputFormat;

import java.io.IOException;

/**
 * Date:2024/10/15
 * Author：Ys
 * Description:
 */
public class WriteSeqFileDriver {

    public static void main(String[] args) throws IOException, InterruptedException, ClassNotFoundException {

        //1.用于管理当前程序所有配置对象
        Configuration conf = new Configuration();
        //2.构建Job对象
        Job job = Job.getInstance(conf, WriteSeqFileDriver.class.getName());
        //3.设置Job的运行主类
        job.setJarByClass(WriteSeqFileDriver.class);
        //4.设置Job的输入为TextInputFormat(普通文本)
        job.setInputFormatClass(TextInputFormat.class);
        //5.设置Job的输入路径
        TextInputFormat.addInputPath(job,new Path("input/file"));
        //6.设置Map端的实现类
        job.setMapperClass(WriteSeqMapper.class);
        //7.设置Map端的输出的key和value类型
        job.setMapOutputKeyClass(NullWritable.class);
        job.setMapOutputValueClass(Text.class);
        //8.设置Job输出为SequenceFile
        job.setOutputFormatClass(SequenceFileOutputFormat.class);
        //9.设置压缩类型为块压缩
        SequenceFileOutputFormat.setOutputCompressionType(job, SequenceFile.CompressionType.BLOCK);
        //10.设置Reduce端的实现类
        job.setReducerClass(WriteSeqReducer.class);
        //11.设置Reduce端的输出的key和value类型
        job.setOutputKeyClass(NullWritable.class);
        job.setOutputValueClass(Text.class);
        //12.设置输出路径
        Path path = new Path("seq_write");
        //13.如果输出路径已经存在就删除
        path.getFileSystem(conf).delete(path,true);
        //14.设置输出路径
        FileOutputFormat.setOutputPath(job,path);
        //15.关闭程序
        boolean result = job.waitForCompletion(true);
        System.exit(result ? 0:1);
    }

}
